package com.headwind.async;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * @author : Headwind
 * @version : 1.0
 * @date : Created in 2020/8/8
 * @description : 使用CompletableFuture实现异步化
 */
public class CompletableFutureSamples {

    /**
     * 获取处理器数量
     */
    private final static int CPUS = Runtime.getRuntime().availableProcessors();
    /**
     * 初始化线程池
     */
    private final static ThreadPoolExecutor THREAD_POOL_EXECUTOR =
            new ThreadPoolExecutor(CPUS, CPUS * 2, 1, TimeUnit.MINUTES,
                    new LinkedBlockingQueue<>(5), new ThreadPoolExecutor.CallerRunsPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
//        completableFuture();
//        runAsync();
//        supplyAsync();
//        thenRun();
//        thenAccept();
//        thenApply();
//        whenComplete();

        // 基于thenCompose
//        CompletableFuture completableFuture1 = task1("A123")
//                .thenCompose(id -> task2(id));
//        System.out.println(completableFuture1.get());

        // 基于thenCombine
//        CompletableFuture completableFuture2 = task1("A123").thenCombine(task2("B123"),
//                (one, two) -> {
//                    return one + " " + two;
//                });
//        System.out.println(completableFuture2.get());

//        allOf();
//        anyOf();
        streamAndCompletableFuture();
    }

    /**
     * CompletableFuture实现异步
     */
    public static void completableFuture() throws ExecutionException, InterruptedException, TimeoutException {
        long start = System.currentTimeMillis();
        // 创建 CompletableFuture 任务
        CompletableFuture<String> completableFuture = new CompletableFuture<String>();
        // 使用线程池来开启线程计算任务结果
        THREAD_POOL_EXECUTOR.execute(() -> {
            // 休眠2秒，模拟任务计算
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 设置计算结果
            System.out.println("----" + Thread.currentThread().getName() + "设置输出结果----");
            completableFuture.complete("Hello CompletableFuture!");
        });
        // 等待计算结果
        System.out.println("----主线程等待 completableFuture 的计算结果----");
        // 输出两个任务的执行结果
        System.out.println(completableFuture.get());
        System.out.println(completableFuture.get(1000, TimeUnit.MICROSECONDS));
        System.out.println("----主线程获取到了 completableFuture 的计算结果----");
        THREAD_POOL_EXECUTOR.shutdown();
    }

    /**
     * CompletableFuture的runAsync方法
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void runAsync() throws ExecutionException, InterruptedException {
        // 创建异步任务，并返回Future
        CompletableFuture completableFuture = CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
                // 休眠2秒模拟任务
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("----任务执行结束----");
            }
        });
        // 同步等待异步任务执行结束返回<null>
        System.out.println(completableFuture.get());
    }

    /**
     * CompletableFuture的supplyAsync方法
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void supplyAsync() throws ExecutionException, InterruptedException {
        // 创建异步任务，并返回Future
        CompletableFuture completableFuture = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                // 休眠2秒模拟任务
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 返回异步计算结果
                return "Hello SupplyAsync";
            }
        });
        // 同步等待异步任务执行结束返回
        System.out.println(completableFuture.get());
    }

    /**
     * CompletableFuture的thenRun方法
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void thenRun() throws ExecutionException, InterruptedException {
        // 创建异步任务，并返回Future
        CompletableFuture<String> completableFutureA = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                // 休眠2秒模拟任务
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 返回异步计算结果
                return "Hello completableFutureA";
            }
        });
        // 在completableFutureA上施加事件，当completableFutureA计算完成后回调该事件并返回新Future
        CompletableFuture completableFutureB = completableFutureA.thenRun(new Runnable() {
            @Override
            public void run() {
                // 当任务A完了执行该任务
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName());
                System.out.println("----在任务A结束后执行该任务----");
            }
        });
        // 同步等待任务B执行结束返回<null>
        System.out.println(completableFutureB.get());
    }

    /**
     * CompletableFuture的thenAccept方法
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void thenAccept() throws ExecutionException, InterruptedException {
        // 创建异步任务，并返回Future
        CompletableFuture<String> completableFutureA = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                // 休眠2秒模拟任务
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 返回异步计算结果
                return "Hello completableFutureA";
            }
        });
        // 在completableFutureA上施加事件，当completableFutureA计算完成后回调该事件并返回新Future
        CompletableFuture completableFutureB = completableFutureA.thenAccept(new Consumer<String>() {
            @Override
            public void accept(String s) {
                // 当任务A完了执行该任务
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName());
                System.out.println("----在任务A结束后执行该任务----" + s);
            }
        });
        // 同步等待任务B执行结束返回<null>
        System.out.println(completableFutureB.get());
    }

    /**
     * CompletableFuture的thenApply方法
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void thenApply() throws ExecutionException, InterruptedException {
        // 创建异步任务，并返回Future
        CompletableFuture<String> completableFutureA = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                // 休眠2秒模拟任务
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 返回异步计算结果
                return "Hello completableFutureA";
            }
        });
        // 在completableFutureA上施加事件，当completableFutureA计算完成后回调该事件并返回新Future
        CompletableFuture completableFutureB = completableFutureA.thenApply(new Function<String, String>() {

            @Override
            public String apply(String s) {
                // 当任务A完了执行该任务
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return s + " completableFutureB";
            }
        });
        // 同步等待任务B执行结束返回
        System.out.println(completableFutureB.get());
    }

    /**
     * CompletableFuture的whenComplete方法
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void whenComplete() throws ExecutionException, InterruptedException {
        // 创建异步任务，并返回Future
        CompletableFuture<String> completableFutureA = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                // 休眠2秒模拟任务
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 返回异步计算结果
                return "Hello completableFutureA";
            }
        });
        // 添加回调函数
        completableFutureA.whenComplete(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String s, Throwable throwable) {
                // 如果没有异常，打印异步任务结果
                if (null == throwable) {
                    System.out.println(s);
                } else {
                    // 打印异常信息
                    System.out.println(throwable.getLocalizedMessage());
                }
            }
        });
        // 挂起当前线程，等待异步任务执行完毕
        Thread.currentThread().join(2000);
    }

    /**
     * 异步任务返回future
     *
     * @param taskId 任务Id
     * @return 执行结果
     */
    public static CompletableFuture<String> task1(String taskId) {
        // 异步任务，返回Future
        return CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                // 休眠1秒模拟任务
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 返回异步计算结果
                return taskId;
            }
        });
    }

    /**
     * 异步任务返回future
     *
     * @param taskId 任务Id
     * @return 执行结果
     */
    public static CompletableFuture<String> task2(String taskId) {
        // 异步任务，返回Future
        return CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                // 休眠3秒模拟任务
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 返回异步计算结果
                return taskId + ":ali";
            }
        });
    }

    /**
     * 基于allOf实现多个转一个
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void allOf() throws ExecutionException, InterruptedException {
        // 创建future列表
        List<CompletableFuture<String>> completableFutureList = new ArrayList<>();
        completableFutureList.add(task1("1"));
        completableFutureList.add(task1("2"));
        completableFutureList.add(task1("3"));
        completableFutureList.add(task1("4"));
        // 转换多个future为一个
        CompletableFuture<Void> result = CompletableFuture.allOf(completableFutureList.toArray(
                new CompletableFuture[completableFutureList.size()]));
        // 等待结果返回
        System.out.println(result.get());

    }

    /**
     * 基于anyOf实现多个转一个
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void anyOf() throws ExecutionException, InterruptedException {
        // 创建future列表
        List<CompletableFuture<String>> completableFutureList = new ArrayList<>();
        completableFutureList.add(task1("1"));
        completableFutureList.add(task1("2"));
        completableFutureList.add(task2("3"));
        // 转换多个future为一个
        CompletableFuture<Object> result = CompletableFuture.anyOf(completableFutureList.toArray(
                new CompletableFuture[completableFutureList.size()]));
        // 等待结果返回
        System.out.println(result.get());

    }

    /**
     * Stream结合CompletableFuture实现异步流式处理
     */
    public static void streamAndCompletableFuture() {
        // 生成IP列表
        List<String> ipList = new ArrayList<>();
        for (int i = 1; i <= 10; i++) {
            ipList.add("192.168.12." + i);
        }
        // 并发调用
        long start = System.currentTimeMillis();
        List<CompletableFuture<String>> completableFutureList = ipList.stream()
                //同步转换为异步
                .map(ip -> CompletableFuture.supplyAsync(() -> rpcCall(ip, ip)))
                //收集结果
                .collect(Collectors.toList());
        //等待所有异步任务执行完成
        List<String> resultList = completableFutureList.stream()
                //同步等待结果
                .map(future -> future.join())
                //对结果进行收集
                .collect(Collectors.toList());
        //输出
        resultList.stream().forEach(r -> System.out.println(r));
        System.out.println("总耗时：" + (System.currentTimeMillis() - start));
    }

    /**
     * 模拟远程调用
     *
     * @param ip    IP地址
     * @param param 参数
     * @return 参数
     */
    public static String rpcCall(String ip, String param) {
        System.out.println(ip + "rpcCall:" + param);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return param;
    }
}
